package org.zhaya.sdk.processors;

import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.logz.sawmill.Doc;
import io.logz.sawmill.ProcessResult;
import io.logz.sawmill.Processor;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.exceptions.ProcessorConfigurationException;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.zhaya.sdk.DodpProcessor;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @Author Ryan Yang
 * @Date 2022/10/17 7:00 下午
 */
@ProcessorProvider(type = "extractField", factory = ExtractFieldProcessor.Factory.class)
public class ExtractFieldProcessor implements Processor {
    private final List<String> fieldList;

    public ExtractFieldProcessor(List<String> fieldList) {
        this.fieldList = fieldList;
    }

    /**
     * @return {@link ProcessResult}
     * @description: doc只保留需要的字段
     */
    @Override
    public ProcessResult process(Doc doc) throws InterruptedException {

        Map<String, Object> collect = Maps.newHashMap();
        Map<String, Object> source = doc.getSource();
        for (String field : fieldList) {
            if (doc.hasField(field)) {
                collect.put(field, doc.getField(field));
            }
        }
        for (String s : new ArrayList<>(source.keySet())) {
            doc.removeField(s.replace(StrUtil.DOT, "\\."));
        }
        for (String field : fieldList) {
            doc.addField(field, collect.get(field));
        }
        return ProcessResult.success();
    }

    /**
     * @description: 构建组件处理器
     */
    public static class Factory implements DodpProcessor.Factory {
        @Override
        public Processor create(Map<String, Object> config) {
            Optional<Object> fieldOpt = Optional.ofNullable(config.get("fieldList"));
            if (!fieldOpt.isPresent()) {
                throw new ProcessorConfigurationException("Processor conf is error!");
            }
            List<String> collect = (List<String>) (fieldOpt.get() instanceof Collection ?
                    fieldOpt.get() : Lists.newArrayList(fieldOpt.get().toString().split(StrUtil.COMMA)));
            List<String> duplicateElements = getDuplicateElements(collect.stream());
            if (duplicateElements.size() > 0) {
                throw new ProcessorConfigurationException(
                        String.format("fieldList has repeated fields %s", duplicateElements));
            }
            return new ExtractFieldProcessor(collect);
        }
    }

    public static <T> List<T> getDuplicateElements(Stream<T> stream) {
        return stream
                .collect(Collectors.toMap(e -> e, e -> 1, Integer::sum))
                .entrySet().stream()
                .filter(entry -> entry.getValue() > 1)
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Configuration implements Processor.Configuration {
        private List<String> fieldList;
    }

}
